package com.atguigu.edu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
//todo 1 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 1000L, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 *1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");
        env.setStateBackend(new HashMapStateBackend());
        */

        //todo 3 读取kafka主题topic_log中的数据
        String topicName = "topic_log";
        String groupId = "base_log_app";
        DataStreamSource<String> topicDbStream = env.addSource(
                KafkaUtil.getKafkaConsumer(topicName, groupId));
        //topicDbStream.print("base_log_app>>>>>>>>>>>>>>>");
        //todo 4 清洗转换数据
        OutputTag<String> dirtyOutputTag = new OutputTag<String>("Dirty") {
        };
        SingleOutputStreamOperator<JSONObject> jsonObjectStream = topicDbStream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    e.printStackTrace();
                    ctx.output(dirtyOutputTag, value);
                }
            }
        });

        //todo 5 根据mid分组数据
        KeyedStream<JSONObject, String> keyedStream = jsonObjectStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                String string = value.getJSONObject("common").getString("mid");
                return string;
            }
        });
//todo 6 根据键控状态完成新旧访客的标记修复
        SingleOutputStreamOperator<JSONObject> isNewStream = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            //装填的格式 yyyy-MM-dd
            private ValueState<String> firstVisitDtState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                firstVisitDtState = getRuntimeContext()
                        .getState(new ValueStateDescriptor<String>("first_visit_dt", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                String isNew = value.getJSONObject("common").getString("is_new");
                String firstVisitDt = firstVisitDtState.value();
                String visitDt = DateFormatUtil.toDate(value.getLong("ts"));
                if ("1".equals(isNew)) {
                    if (firstVisitDt == null) {
                        //没有登录过的新访客
                        firstVisitDtState.update(visitDt);
                    } else if (!firstVisitDt.equals(visitDt)) {
                        //删除了app的旧访客 不是新访客
                        value.getJSONObject("common").put("is_new", "0");
                    }
                } else {
                    //标记为零
                    if (firstVisitDt == null) {
                        //修改状态日期为前1天
                        firstVisitDtState.update(DateFormatUtil.toDate(value.getLong("ts") - 1000L * 60 * 60 * 24));
                    }
                }
                return value;
            }
        });

        //todo 7 将数据拆分为5种数据流
        //page页面做为主流 另外5种start  err  action  display appVideo做为册输出流
        OutputTag<String> startOutputTag = new OutputTag<String>("start"){};
        OutputTag<String> appVideoOutputTag = new OutputTag<String>("appVideo"){};
        OutputTag<String> errOutputTag = new OutputTag<String>("err"){};
        OutputTag<String> actionOutputTag = new OutputTag<String>("action"){};
        OutputTag<String> displayOutputTag = new OutputTag<String>("display"){};
        SingleOutputStreamOperator<String> pageStream = isNewStream
                .process(new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                        //topic_log中有两种数据 start page
                        //1.错误数据输出到侧输出流
                        String err = value.getString("err");
                        if (err != null) {
                            ctx.output(errOutputTag, err);
                            value.remove("err");
                        }

                        //2.启动日志和视频日志输出到侧输出流
                        JSONObject start = value.getJSONObject("start");
                        JSONObject appVideo = value.getJSONObject("appVideo");
                        //获取公共信息
                        JSONObject common = value.getJSONObject("common");
                        Long ts = value.getLong("ts");
                        JSONObject page = value.getJSONObject("page");
                        if (start != null) {
                            //当前为启动日志  将全部数据写出到start侧输出流
                            ctx.output(startOutputTag, start.toJSONString());
                            value.remove("start");
                        } else if(appVideo != null){
                            //当前为视频播放日志  将全部数据写出到appVideo侧输出流
                            appVideo.put("common", common);
                            appVideo.put("ts", ts);
                            appVideo.put("page", page);
                            ctx.output(appVideoOutputTag, appVideo.toJSONString());
                            value.remove("appVideo");
                        }else {
                            //当前页面日志
                            //3.打散输出action数据
                            JSONArray actions = value.getJSONArray("actions");
                            if (actions != null) {
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    action.put("common", common);
                                    action.put("ts", ts);
                                    action.put("page", page);
                                    ctx.output(actionOutputTag, action.toJSONString());
                                }

                            }
                            value.remove("actions");
                            //4.打散输出display数据
                            JSONArray displays = value.getJSONArray("displays");
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    display.put("common", common);
                                    display.put("ts", ts);
                                    display.put("page", page);
                                    ctx.output(displayOutputTag, display.toJSONString());
                                }
                                value.remove("displays");
                            }

                            //5.主流输出page数据
                            out.collect(value.toJSONString());
                        }
                    }
                });

        DataStream<String> appVideoStream = pageStream.getSideOutput(appVideoOutputTag);
        DataStream<String> startStream = pageStream.getSideOutput(startOutputTag);
        DataStream<String> errStream = pageStream.getSideOutput(errOutputTag);
        DataStream<String> actionStream = pageStream.getSideOutput(actionOutputTag);
        DataStream<String> displayStream = pageStream.getSideOutput(displayOutputTag);

        //pageStream.print("pageStream >>>>>");
        //startStream.print("startStream >>>>>");
        //errStream.print("errStream >>>>>");
        //actionStream.print("actionStream >>>>>");
        //displayStream.print("displayStream >>>>>");
        //todo 8 将6种数据流写回到对应的kafka主题
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        String appVideo_topic = "dwd_traffic_app_Video";

        appVideoStream.addSink(KafkaUtil.getKafkaProducer(appVideo_topic));
        pageStream.addSink(KafkaUtil.getKafkaProducer(page_topic));
        startStream.addSink(KafkaUtil.getKafkaProducer(start_topic));
        displayStream.addSink(KafkaUtil.getKafkaProducer(display_topic));
        actionStream.addSink(KafkaUtil.getKafkaProducer(action_topic));
        errStream.addSink(KafkaUtil.getKafkaProducer(error_topic));

        env.execute(groupId);
    }
}
